package com.dayu.finecomm.device;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.dayu.finecomm.decoder.HdTcpDecoder;
import com.dayu.finecomm.handler.tpacket.TPacketCollectHandler;
import com.dayu.finecomm.handler.tpacket.TPacketCommitHandler;
import com.dayu.finecomm.handler.tpacket.TPacketOPCUAHandler;
import com.dayu.finecomm.mongo.pojo.IotPlat;
import com.dayu.finecomm.mongo.service.IotPlatService;
import com.dayu.finecomm.pojo.DeviceConfig;
import com.dayu.finecomm.pojo.DevicePusher;
import com.dayu.finecomm.protocol.Protocols;
import com.dayu.finecomm.protocol.SpecialCareChannelInboundAdapter;
import com.dayu.finecomm.protocol.codec.Drive;
import com.dayu.finecomm.protocol.codec.ProtocolDecoder;
import com.dayu.finecomm.protocol.log.ProtocolLogger;
import com.dayu.finecomm.protocol.pojo.TPacket;
import com.dayu.finecomm.source.CollectSource;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * Iot设备驱动
 * 
 * @author XuJialiang
 * @since 2017-09-09 14:22:12
 * @version V1.0.0
 */
@Drive(IotPlatDevice.DRIVE)
public class IotPlatDevice extends ChannelInitializer<NioSocketChannel> implements Device {

	static final String DRIVE = "iotplat";

	static Logger LOGGER = LoggerFactory.getLogger(IotPlatDevice.class);

	/**
	 * Channel Handler Listeners
	 */
	private Map<String, ChannelHandlerContext> listeners;

	/**
	 * 协议 NB-IOT
	 */
	final Set<String> protocols;
	private Set<InetSocketAddress> distributors;
	private Collection<DevicePusher> pushers;

	final DeviceConfig config;
	final String prefix;
	private CollectSource stations;
	private String listenType;

	/**
	 * NB-IOT解码器
	 */
	private List<ProtocolDecoder<TPacket>> protocolDecoders;
	Map<String, byte[]> IDs;

	/**
	 * MongoDB Server
	 */
	private IotPlatService dbService;

	public IotPlatDevice(CollectSource stations, Set<String> protocols, DeviceConfig config,
			IotPlatService iotplatService) {

		this.protocols = protocols;
		this.prefix = config.prefix;
		this.stations = stations;
		this.config = config;

		protocolDecoders = Protocols.protocols(this.protocols);
		IDs = new ConcurrentHashMap<>();

		this.dbService = iotplatService;
	}

	@Override
	protected void initChannel(NioSocketChannel ctx) throws Exception {

		ChannelPipeline pipeline = ctx.pipeline();

		pipeline.addLast(new IotDataHandler());
		pipeline.addLast(new HdTcpDecoder(this.prefix, protocolDecoders));
		pipeline.addLast(new TPacketCommitHandler(this));
		pipeline.addLast(new TPacketCollectHandler(stations, false, config.name));

		if (config.opcUaEnabled) {
			pipeline.addLast(new TPacketOPCUAHandler(config.opcUaEndPoint));
		}

		// pipeline.addLast(new CMDResultHandler(stations, this));
		// pipeline.addLast(new CMDStoreHandler(stations));

		pipeline.addLast(new ProtocolLogger());
	}

	/**
	 * 处理IOT发送过来的数据，并交由IOT协议进行Decoder
	 * 
	 * @author XuJialiang
	 * @since 2017-09-09 14:22:12
	 * @version V1.0.0
	 */
	class IotDataHandler extends SpecialCareChannelInboundAdapter<JSONObject> {

		@Override
		protected void channelRead0(ChannelHandlerContext ctx, JSONObject content) throws Exception {
			String subscribe = content.getString("subscribe");

			if (0 == subscribe.compareTo("deviceDatasChanged") || 0 == subscribe.compareTo("commandRsp")) {
				String appid = content.getString("appId");
				IotPlat iotplat = dbService.query(appid);

				if (null != iotplat) {
					content.put("protocol", iotplat.getProtocol());

					if (0 == subscribe.compareTo("deviceDatasChanged")) {
						// 如有需要，进行切包处理
						ByteBuf heapBuffer = Unpooled.buffer();
						heapBuffer.writeBytes(content.toString().getBytes());
						ctx.fireChannelRead(heapBuffer);
					} else {
						ByteBuf heapBuffer = Unpooled.buffer();
						heapBuffer.writeBytes(content.toString().getBytes());
						ctx.fireChannelRead(heapBuffer);
					}
				} else {
					LOGGER.error("mongoDB is offline, appid=" + appid);
				}
			} else {
				LOGGER.info("currentlly, not supprt subscribe " + subscribe);
			}
		}

	}

	@Override
	public Collection<DevicePusher> getPushers() {
		return pushers;
	}

	@Override
	public void setPushers(Collection<DevicePusher> pushers) {
		this.pushers = pushers;
	}

	@Override
	public Map<String, ChannelHandlerContext> getListeners() {
		return listeners;
	}

	@Override
	public void setListeners(Map<String, ChannelHandlerContext> lITENERS) {
		listeners = lITENERS;
	}

	@Override
	public Set<InetSocketAddress> getDistributors() {
		return distributors;
	}

	@Override
	public void setDistributors(Set<InetSocketAddress> distributors) {
		this.distributors = distributors;
	}

	@Override
	public String getListenType() {
		return listenType;
	}

	@Override
	public void setListenType(String listenType) {
		this.listenType = listenType;
	}
}
